package hbasecli.hbase;

import hbasecli.config.ConnectionConfig;
import hbasecli.exception.DaoException;
import hbasecli.param.ListTable;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FuzzyRowFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.stream.Collectors;

import static cn.hutool.core.text.CharSequenceUtil.cleanBlank;

@SuppressWarnings("all")
@Slf4j
@Component
public class HbaseDao {
    private static Connection connection;

    @Resource
    private ConnectionConfig connectionConfig;

    @PostConstruct
    private void init() {
        if (connection != null) {
            return;
        }
        connection = createConnection();
    }

    public Connection createConnection() {
        log.info(connectionConfig.toString());
        Configuration configuration = new Configuration();
        configuration.set("hbase.zookeeper.property.clientPort", connectionConfig.getZookeeperPropertyClientPort());
        configuration.set("hbase.zookeeper.quorum", connectionConfig.getZookeeperQuorum());
        configuration.set("hbase.master.port", connectionConfig.getMasterPort());
        configuration.set("hbase.regionserver.port", connectionConfig.getRegionserverPort());
        Connection connection = null;
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
            throw new DaoException("连接时出现异常:" + e.getMessage());
        }
        if (connection == null) {
            throw new DaoException("创建连接失败");
        }
        log.info("HBase 连接成功");
        log.info(connectionConfig.toString());
        return connection;
    }

    public List<HbaseTable> listTable(ListTable param) {
        try {
            Admin admin = connection.getAdmin();
            List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
            if (tableDescriptors == null) {
                return Collections.emptyList();
            } else {
                List<HbaseTable> hbaseTables = new ArrayList<>();
                for (TableDescriptor td : tableDescriptors) {
                    HbaseTable hbaseTable = new HbaseTable();
                    hbaseTable.setName(td.getTableName().toString());
                    List<String> families = new ArrayList<>();
                    Map<String, Map<String, String>> familyInfo = new HashMap<>();
                    for (ColumnFamilyDescriptor cfd : td.getColumnFamilies()) {
                        String family = cfd.getNameAsString();
                        families.add(family);
                        Map<String, String> info = new HashMap<>();
                        Map<Bytes, Bytes> values = cfd.getValues();
                        for (Bytes bytes : values.keySet()) {
                            info.put(bytes.toString(), values.get(bytes).toString());
                        }
                        familyInfo.put(family, info);
                    }
                    hbaseTable.setFamilyInfo(familyInfo);
                    hbaseTable.setFamilies(families);
                    hbaseTables.add(hbaseTable);
                }
                return hbaseTables;
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }


    public boolean createTable(String tableName, List<String> families, boolean deleteIfExists) {
        try {
            Admin admin = connection.getAdmin();
            TableDescriptorBuilder builder =
                    TableDescriptorBuilder.newBuilder(TableName.valueOf(cleanBlank(tableName)));
            List<ColumnFamilyDescriptor> columnFamilies = new ArrayList<>();
            for (String familyName : families) {
                columnFamilies.add(
                        ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cleanBlank(familyName))).build()
                );
            }
            builder.setColumnFamilies(columnFamilies);
            TableDescriptor table = builder.build();
            if (deleteIfExists && admin.tableExists(table.getTableName())) {
                admin.disableTable(table.getTableName());
                admin.deleteTable(table.getTableName());
            }
            admin.createTable(table);
            admin.close();
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }

    public boolean deleteTable(String tableName) {
        try {
            Admin admin = connection.getAdmin();
            TableName name = TableName.valueOf(cleanBlank(tableName));
            if (!admin.tableExists(name)) {
                return true;
            }
            admin.disableTable(name);
            admin.deleteTable(name);
            admin.close();
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }

    public boolean deleteTable(List<String> tableNames) {
        boolean ok = true;
        for (String tableName : tableNames) {
            ok &= deleteTable(cleanBlank(tableName));
        }
        return ok;
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, byte[] bytesValue) {
        try {
            Table table = connection.getTable(TableName.valueOf(cleanBlank(tableName)));
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes(cleanBlank(family)), Bytes.toBytes(cleanBlank(qualifier)), bytesValue);
            table.put(put);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, BigDecimal bigDecimalValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(bigDecimalValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, boolean boolValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(boolValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, ByteBuffer byteBufferValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(byteBufferValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, double doubleValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(doubleValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, float floatValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(floatValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, int intValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(intValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, long longValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(longValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, short shortValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(shortValue));
    }

    public boolean put(String tableName, byte[] rowKey, String family, String qualifier, String stringValue) {
        return put(tableName, rowKey, family, qualifier, Bytes.toBytes(stringValue));
    }

    public List<HbaseRecord> scan(String tableName, Filter filter, String startRow) {
        try {
            List<HbaseRecord> resultList = new ArrayList<>();
            Table table = connection.getTable(TableName.valueOf(cleanBlank(tableName)));
            Scan scan = new Scan();
            if (startRow != null) {
                scan.withStartRow(Bytes.toBytes(startRow));
            }
            if (filter != null) {
                scan.setFilter(filter);
            }
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                HbaseRecord hbaseRecord = new HbaseRecord();
                List<HbaseRecord.HbaseRecordColumn> columns = new ArrayList<>();
                for (Cell cell : result.listCells()) {
                    hbaseRecord.setRowKey(CellUtil.cloneRow(cell));
                    HbaseRecord.HbaseRecordColumn column = new HbaseRecord.HbaseRecordColumn();
                    column.setFamily(Bytes.toString(CellUtil.cloneFamily(cell)));
                    column.setQualifier(Bytes.toString(CellUtil.cloneQualifier(cell)));
                    column.setDataType("");
                    column.setValue(CellUtil.cloneValue(cell));
                    columns.add(column);
                }
                hbaseRecord.setColumns(columns);
                resultList.add(hbaseRecord);
            }
            return resultList;
        } catch (Exception e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }

    public List<HbaseRecord> scan(String tableName, Filter filter) {
        return scan(tableName, filter, null);
    }

    public List<HbaseRecord> scan(String tableName) {
        return scan(tableName, null);
    }

    public List<HbaseRecord> scanWithPrefix(String tableName, String rowPrefix, String startRow) {
        return scan(tableName, new PrefixFilter(Bytes.toBytes(rowPrefix)), startRow);
    }

    public List<HbaseRecord> scanWithPrefix(String tableName, String rowPrefix) {
        return scan(tableName, new PrefixFilter(Bytes.toBytes(rowPrefix)));
    }

    public List<HbaseRecord> fuzzyScan(String tableName, String rowKeyPattern, char placeholder, String startRow) {
        byte[] bytes = Bytes.toBytesBinary(rowKeyPattern);
        byte[] mask = new byte[bytes.length];
        for (int i = 0; i < bytes.length; i++) {
            if (bytes[i] == placeholder) {
                mask[i] = (byte) 1;
            }
        }
        Pair<byte[], byte[]> pair = new Pair<>(bytes, mask);
        List<Pair<byte[], byte[]>> fuzzyKeysData = Arrays.asList(pair);
        FuzzyRowFilter filter = new FuzzyRowFilter(fuzzyKeysData);
        return scan(tableName, filter, startRow);
    }

    public List<HbaseRecord> fuzzyScan(String tableName, String rowKeyPattern, char placeholder) {
        return fuzzyScan(tableName, rowKeyPattern, placeholder, null);
    }

    public List<HbaseRecord> fuzzyScan(String tableName, String rowKeyPattern, String startRow) {
        return fuzzyScan(tableName, rowKeyPattern, '?', startRow);
    }

    public List<HbaseRecord> fuzzyScan(String tableName, String rowKeyPattern) {
        return fuzzyScan(tableName, rowKeyPattern, '?', null);
    }

    public boolean deleteRecord(String tableName, List<byte[]> rowKeys) {
        try {
            Table table = connection.getTable(TableName.valueOf(cleanBlank(tableName)));
            List<Delete> deletes = rowKeys.stream()
                    .map(rk -> new Delete(rk))
                    .collect(Collectors.toList());
            table.delete(deletes);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            throw new DaoException("数据访问异常:" + e.getMessage());
        }
    }
}
